package com.xj.center.data.service.kafka;

import cn.hutool.core.thread.ThreadUtil;
import com.xj.center.data.service.ControlInfoService;
import com.xj.center.data.service.DataCustomService;
import com.xj.center.data.service.PointValueService;
import com.xj.center.data.service.job.PointValueScheduleJob;
import com.xj.common.bean.constant.Common;
import com.xj.common.bean.em.ControlStateEm;
import com.xj.common.bean.point.PointValue;
import com.xj.common.model.ControlInfo;
import com.xj.common.model.DriverEvent;
import com.xj.common.utils.JsonPaseUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
* 类说明 
*/
@Slf4j
@Component
public class ControlInfoConsumer {
    public static Map<String,Integer> mapReturnCode = new HashMap();
    @Resource
    private ControlInfoService controlInfoService;
    @Resource
    private ThreadPoolExecutor threadPoolExecutor;
    // 消费监听，接收驱动层发送过来的消息
    @KafkaListener(topics = {Common.KAFKA.TOPIC_EXCHANGE_CONTROL_EVENT},groupId = "${ControlInfoGroup}")
    public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack){
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("ControlInfoConsumer："+record.topic()+"-"+record.partition()+"-"+record.value());
        String jsonstr = record.value().toString();
        ControlInfo controlInfo = JsonPaseUtil.JSONStrToObj(jsonstr, ControlInfo.class);

        if (null == controlInfo || null == controlInfo.getDeviceId()) {
            log.error("Invalid controlInfo value: {}", controlInfo);
            return;
        }        String msgType = controlInfo.getMsgType();

        switch (msgType){
            case Common.Device.ControlEvent.REQUEST : { doRequest(controlInfo); break;  }//控制请求物联网处理，生成一条新的记录
            case Common.Device.ControlEvent.RESPONSE : { doResponse(controlInfo); break;  }//修改记录，下控令成功
            case Common.Device.ControlEvent.CTLERROR : { doCtleError(controlInfo); break;  }//生成一条记录，记录错误原因
            default:{
                log.info("暂不识别的控制处理类型操作",msgType);
            }

        }

        //提交偏移量
        ack.acknowledge();
    }
    private void                                                                                                                                                      doRequest(ControlInfo controlInfo){
        mapReturnCode.put(controlInfo.getId(),ControlStateEm.CTL_STATE_RECEIVED.getCode());

        controlInfo.setResultcode(ControlStateEm.CTL_STATE_RECEIVED.getCode());
        controlInfo.setResultmsg(ControlStateEm.CTL_STATE_RECEIVED.getMsg());
        controlInfoService.saveControlInfoToRedis(controlInfo);
        boolean flag = true;
        //开启一个定时超时任务，超时则直接报错
        try {
            threadPoolExecutor.submit(() -> {
                //超时的判定，每一秒查一次直到正常结束   getResuCodeCh(controlInfo)
                while (mapReturnCode.get(controlInfo.getId()) == ControlStateEm.CTL_STATE_RECEIVED.getCode()) {

                    ThreadUtil.sleep(1000);

                }
            }).get(controlInfo.getTimeOut(), TimeUnit.SECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException exception) {
            //进行超时操作
            doTimeout(controlInfo);
        }




    }


    private void doResponse(ControlInfo controlInfo){
        try {

            //获取redis中的控制缓存数据
            ControlInfo controlInfoR = controlInfoService.realtimeByCtlId(controlInfo.getId());
            //更新对应数据
            controlInfoR.setResultcode(ControlStateEm.CTL_STATE_RESPONSE.getCode());
            controlInfoR.setResultmsg(ControlStateEm.CTL_STATE_RESPONSE.getMsg());
            controlInfoService.updateControlInfoToRedis(controlInfoR);
            //写入mongo数据库,控制结束
            controlInfoService.saveControlInfoToMongo(controlInfo);
            //更新缓存，用来关闭超时判定线程
            mapReturnCode.replace(controlInfo.getId(),ControlStateEm.CTL_STATE_RESPONSE.getCode());
        }catch (Exception e){
            return ;
        }
    }

    /**
     * 直接写入结果
     * @param controlInfo
     */
    private void doCtleError(ControlInfo controlInfo){
        /*//获取redis中的控制缓存数据
        ControlInfo controlInfoR =  controlInfoService.realtimeByCtlId(controlInfo.getId());*/
        //更新对应数据
        controlInfo.setResultcode(ControlStateEm.CTL_STATE_ERROR.getCode());
        controlInfo.setResultmsg(ControlStateEm.CTL_STATE_ERROR.getMsg());
        //controlInfoService.updateControlInfoToRedis(controlInfoR);
        controlInfo.setTimeUnit(TimeUnit.MINUTES);
        controlInfoService.saveControlInfoToRedis(controlInfo);
        //写入mongo数据库,控制结束
        controlInfoService.saveControlInfoToMongo(controlInfo);
    }

    /**
     * 判超时，写入mongo
     * @param controlInfo
     */
    private void doTimeout(ControlInfo controlInfo){
        //获取redis中的控制缓存数据
        try {
            ControlInfo controlInfoR = controlInfoService.realtimeByCtlId(controlInfo.getId());
            //更新对应数据
            controlInfoR.setResultcode(ControlStateEm.CTL_STATE_TIMEOUT.getCode());
            controlInfoR.setResultmsg(ControlStateEm.CTL_STATE_TIMEOUT.getMsg());
            controlInfoService.updateControlInfoToRedis(controlInfoR);
            //写入mongo数据库,控制结束
            controlInfoService.saveControlInfoToMongo(controlInfo);
        }catch (Exception e){
            return ;
        }
    }

}
